package com.erainm.producer;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * Created by erainm on 2021/9/9 8:52 上午.
 *
 * @Description TODO:
 **/
public class MyProducer1 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Map<String, Object> configs = new HashMap<>();
        // 初始连接用到的broker地址
        configs.put("bootstrap.servers", "lg04:9092");
        // 指定key的序列化类
        configs.put("key.serializer", IntegerSerializer.class);
        // 指定value的序列化类
        configs.put("value.serializer", StringSerializer.class);
        configs.put("acks", "1");
        configs.put("compression.type","gzip");//压缩
        configs.put("reties","3");// 重试次数
        KafkaProducer<Integer, String> producer = new KafkaProducer<Integer, String>(configs);

        // 用于设置用户自定义消息头字段
        List<Header> headers = new ArrayList<>();
        headers.add(new RecordHeader("biz.name", "producer.demo".getBytes()));
        for (int i = 0; i < 100; i++) {
            ProducerRecord<Integer, String> record = new ProducerRecord<>("topic_1", 0, i, "hello lg " + i, headers);

            // 同步消息确认
//        Future<RecordMetadata> future = producer.send(record);
//        RecordMetadata metadata = future.get();
//        System.out.println("消息的主题： " + metadata.topic());
//        System.out.println("消息的分区号： " + metadata.partition());
//        System.out.println("消息的偏移量： " + metadata.offset());
            // 异步消息确认
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null) {
                        System.out.println(
                                "主题:" + recordMetadata.topic() + "\n"
                                        + "分区:" + recordMetadata.partition() + "\n"
                                        + "偏移量:" + recordMetadata.offset() + "\n"
                                        + "序列化的key字节:" + recordMetadata.serializedKeySize() + "\n"
                                        + "序列化的value字节:" + recordMetadata.serializedValueSize() + "\n"
                                        + "时间戳:" + recordMetadata.timestamp());
                    }
                }
            });
        }

        producer.close();
    }
}
